package com.ksc.wordcount.rpc.Driver;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import com.ksc.wordcount.driver.DriverEnv;
import com.ksc.wordcount.worker.ExecutorEnv;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;

public class DriverSystem {

    static ActorSystem system; // 静态变量用于保存ActorSystem实例

    // 获取ExecutorSystem的ActorSystem实例
    public static ActorSystem getExecutorSystem() {
        if (system != null) {
            return system;
        }
        Map<String, Object> map = new HashMap<>();

        map.put("akka.actor.provider", "remote"); // 配置Akka远程Actor提供者
        map.put("akka.remote.transport", "akka.remote.netty.NettyRemoteTransport"); // 配置使用NettyRemoteTransport传输
        map.put("akka.remote.netty.tcp.hostname", DriverEnv.host); // 配置远程TCP主机名
        map.put("akka.remote.netty.tcp.port", DriverEnv.port); // 配置远程TCP端口

        Config config = ConfigFactory.parseMap(map).withFallback(ConfigFactory.load()); // 创建配置对象
        system = ActorSystem.create("DriverSystem", config); // 创建ActorSystem实例
        return system;
    }

    /**
     * executorUrl和akka连接的映射
     */
    static Map<String, ActorRef> executorRefs = new HashMap<>(); // 存储executorUrl和ActorRef的映射关系

    // 获取executorUrl对应的ActorRef实例
    public static ActorRef getExecutorRef(String executorUrl) {
        if (executorRefs.get(executorUrl) != null) {
            return executorRefs.get(executorUrl);
        }
        ActorRef driverRef = getExecutorSystem().actorSelection(executorUrl)
                .resolveOne(Duration.ofSeconds(10)).toCompletableFuture().join(); // 解析并获取ActorRef实例
        executorRefs.put(executorUrl, driverRef); // 将executorUrl和ActorRef加入映射
        return executorRefs.get(executorUrl);
    }
}
